package com.captjack.bigdata.storm.topology;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichSpout;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * storm-spout 组件，类似数据生产者
 *
 * @author Jack Sparrow
 * @version 1.0.0
 * @date 2018/4/13 00:03
 * package com.cc.bigdata.storm.topology
 */
public class StreamDataSpout implements IRichSpout {

    /**
     *
     */
    private SpoutOutputCollector spoutOutputCollector;

    /**
     * 拓扑信息
     */
    private TopologyContext topologyContext;

    /**
     * 随机数工具
     */
    private final Random random = new Random();

    private int index = 0;

    /**
     * @param map                  为此spout提供storm配置
     * @param topologyContext      提供有关拓扑中的spout位置，其任务ID，输入和输出信息的完整信息
     * @param spoutOutputCollector 使我们能够发出将由bolts处理的元组
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.topologyContext = topologyContext;
        this.spoutOutputCollector = spoutOutputCollector;
    }

    /**
     * close是在ISpout接口中定义，用于表示Topology停止。
     */
    @Override
    public void close() {

    }

    @Override
    public void activate() {

    }

    @Override
    public void deactivate() {

    }

    /**
     * nextTuple()从与ack()和fail()方法相同的循环中定期调用。它必须释放线程的控制，
     * 当没有工作要做，以便其他方法有机会被调用。因此，nextTuple的第一行检查处理是否
     * 已完成。如果是这样，它应该休眠至少一毫秒，以减少处理器在返回之前的负载
     */
    @Override
    public void nextTuple() {
        if (this.index <= 1000) {
            List<String> mobileNumbers = new ArrayList<>();
            mobileNumbers.add("1234123401");
            mobileNumbers.add("1234123402");
            mobileNumbers.add("1234123403");
            mobileNumbers.add("1234123404");

            int localIdx = 0;
            while (localIdx++ < 100 && this.index++ < 1000) {
                String fromMobileNumber = mobileNumbers.get(random.nextInt(4));
                String toMobileNumber = mobileNumbers.get(random.nextInt(4));

                while (fromMobileNumber.equals(toMobileNumber)) {
                    toMobileNumber = mobileNumbers.get(random.nextInt(4));
                }

                int duration = random.nextInt(60);
                this.spoutOutputCollector.emit(new Values(fromMobileNumber, toMobileNumber, duration));
            }
        }
    }

    /**
     * ack是在ISpout接口中定义，用于表示Tuple处理成功。
     *
     * @param o 处理对象
     */
    @Override
    public void ack(Object o) {

    }

    /**
     * fail是在ISpout接口中定义，用于表示Tuple处理失败。
     *
     * @param o 处理对象
     */
    @Override
    public void fail(Object o) {

    }

    /**
     * declareOutputFields是在IComponent接口中定义，用于声明数据格式。
     * 即输出的一个Tuple中，包含几个字段。
     *
     * @param outputFieldsDeclarer 推送数据的格式
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("from", "to", "duration"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }

}
